package com.webPV;

import com.weblogpv.WebLogPVMapper;
import com.weblogpv.WebLogPvReducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.net.URI;



public class webMain extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {

        //2、创建job
        Job job = Job.getInstance( this.getConf(), "WebLogUVMapReduce" );
        //设置job运行的主类
        job.setJarByClass( com.weblogpv.WebLogPVMapReduce.class);

        //设置Job
        //a、input
        job.setInputFormatClass(TextInputFormat.class);
        Path inputPath = new Path("hdfs://hadoop102:8020/trackinfo/input/trackinfo.txt");
        TextInputFormat.setInputPaths( job, inputPath);

        //b、map
        job.setMapperClass( WebLogPVMapper.class );
        job.setMapOutputKeyClass( Text.class );
        job.setMapOutputValueClass( IntWritable.class );

        //c.partitioner
        job.setNumReduceTasks(1);

        //d、reduce
        job.setReducerClass( WebLogPvReducer.class);
        job.setOutputKeyClass( Text.class  );
        job.setOutputValueClass( IntWritable.class );

        //e、output
        job.setOutputFormatClass(TextOutputFormat.class);
        Path outputPath = new Path("hdfs://hadoop102:8020/trackinfo/output");

        //如果输出目录存在，先删除
        FileSystem hdfs = FileSystem.get(new URI("hdfs://hadoop102:8020"),new Configuration());
        if(hdfs.exists(outputPath)){
            hdfs.delete( outputPath,true );
        }
        TextOutputFormat.setOutputPath( job,outputPath );

        //第四步，提交job
        boolean isSuccess = job.waitForCompletion( true );

        return isSuccess?0:1 ;
    }


    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        ///public static int run(Configuration conf, Tool tool, String[] args)
        try {
            int  status =  ToolRunner.run( configuration,new com.weblogpv.WebLogPVMapReduce(),args );
            System.exit( status );
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}